package org.etlt;

import org.etlt.job.JobContext;
import org.etlt.job.JobExecutor;
import org.etlt.job.ParalleledJobExecutor;
import org.etlt.job.RunMode;

import java.io.File;
import java.io.IOException;

/**
 * 
 * @author juxiaomi
 * @date 2020年10月20日 - 下午5:29:27
 *
 */
public class Main {

    public static final String JOB_INVENTORY = "job.inventory";

    /**
     *
     * @param args
     * args[0] - etlt config directory
     */
    public static void main(String args[]) throws IOException {
        if(args.length < 1)
            throw new IllegalArgumentException("please set the etl config directory or etl job file.");

        File jobFile = new File(args[0]);
        if(jobFile.exists() && jobFile.isDirectory())
            System.setProperty(JOB_INVENTORY, jobFile.getAbsolutePath());
        else if(jobFile.exists() && jobFile.isFile())
            System.setProperty(JOB_INVENTORY, jobFile.getParentFile().getAbsolutePath());
        JobContext context = new JobContext(jobFile);
        if(args.length > 1){
            String runModeSetting  = args[1];
            RunMode runMode = RunMode.valueOf(runModeSetting.toUpperCase());
            context.setRunMode(runMode);
        }
        context.init();
        runAsParallel(context);
    }

    /**
     * run as parallel, many loaders are running within different threads
     * @param context
     */
    protected static void runAsParallel(JobContext context) throws IOException {
        try {
            ParalleledJobExecutor paralleledJobExecutor = new ParalleledJobExecutor(context.getJobSetting().getParallel());
            paralleledJobExecutor.execute(context);
        }finally {
            context.getJobExecution().doPersistence();
        }
    }

    /**
     * run as serial, all loaders are running within a single thread
     * @param context
     */
    protected static void runAsSerial(JobContext context) throws IOException {
        try {
            JobExecutor jobExecutor = new JobExecutor();
            jobExecutor.execute(context);
        }finally {
            context.getJobExecution().doPersistence();
        }
    }

}
